package org.ala.tree.lsm.sstable;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.ala.tree.lsm.IMemTable;
import org.ala.tree.lsm.KV;
import org.ala.tree.lsm.LSMConfig;
import org.ala.tree.lsm.sstable.block.DataBlock;
import org.ala.tree.lsm.sstable.block.IndexBlock;
import org.ala.tree.lsm.sstable.block.MetaBlock;
import org.ala.tree.lsm.sstable.block.TailBlock;
import org.ala.tree.lsm.tools.BloomFilter;
import org.ala.tree.lsm.tools.FileTools;

/**
 * sstable写入工具
 *
 * @author ala
 * @date 2025年5月28日
 */
public class SSTableWriter<V> {
	/**
	 * 	
	 */
	protected SSTableMerger<V> merge = new SSTableMerger<>();
	/**
	 * 	写入线程
	 */
	protected ThreadPoolExecutor writerThreadPool;
	/**
	 * 	
	 */
	public SSTableWriter() {
		int cpu = Runtime.getRuntime().availableProcessors();
		AtomicInteger crt = new AtomicInteger();
		writerThreadPool = new ThreadPoolExecutor(
				cpu, cpu, 
				180, TimeUnit.SECONDS, 
				new ArrayBlockingQueue<>(1024),
				new ThreadFactory() {
					public Thread newThread(Runnable r) {
						Thread t = new Thread(r);
						t.setName(String.format("SSTableWriteThread_%d", crt.getAndIncrement()));
						t.setDaemon(false);
						return t;
					}
				});
	}

	
	/**
	 * 	保存L0层
	 * 	@throws IOException 
	 */
	public void save(LSMConfig config, List<IMemTable<V>> memTables, Class<V> vclass) {
		//	检测目录是否存在
		checkLevelDir(config, 0);
		//	落盘内存块
		//	TODO	这里其实有问题，落盘的时候要写锁。防止一遍落盘一遍合并
		memTables.forEach(mt -> {
			try {
				_save(config, mt);
				//	落盘成功后，清除wal
				mt.clearWal();
				merge.tryMerge(config, 0, vclass, this);
			} catch (IOException e) {
				e.printStackTrace();
			}
		});
	}
	/**
	 * 	MemTable内容落盘
	 * 	@throws IOException 
	 */
	protected void _save(LSMConfig config, IMemTable<V> memTable) throws IOException {
		String path = String.format("%s/%s%d/%s.sstable", config.ROOT_DIR, config.levelPrefix, 0, memTable.id());
		save(path, memTable.size(), config.blockCapacity, memTable.iterator());
	}
	public void save(String path, int size, int blockCapacity, Iterator<KV<V>> it) throws IOException {
		//	打开的永远是新文件，pos指针永远是0
		FileChannel fileChannel = FileChannel.open(Path.of(path), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
		
		//	写入DataBlock, IndexBlock, 布隆过滤器（0.1%的误判率）
		DataBlock<V> db = null;
		IndexBlock<V> ib = new IndexBlock<>();
		BloomFilter bf = new BloomFilter(size, 0.001d);
		KV<V> prev = null;
		int pos = 0;
		for ( ; it.hasNext() ; ) {
			if (db == null) {
				db = new DataBlock<>(blockCapacity);
			}
			
			KV<V> kv = it.next();
			//	去重，保留最新版的数据
			if (prev == null) {
				prev = kv;
			} else if (prev.getKey().equals(kv.getKey())) {
				prev = prev.getTimestamp() > kv.getTimestamp() ? prev : kv;
			} else {
				bf.add(prev.getKey());
				db.append(prev.getKey(), prev.getVal());
				//	如果db写满了，刷盘
				if (db.isFull()) {
					ByteBuffer buf = db.encode();
					ib.append(pos, db);
					pos += buf.capacity();
					fileChannel.write(buf);
					db = null;
				}
				prev = kv;
			}
		}
		if (prev != null) {
			db.append(prev.getKey(), prev.getVal());
		}
		//	把尾巴写掉
		if (db != null) {
			ByteBuffer buf = db.encode();
			ib.append(pos, db);
			pos += buf.capacity();
			fileChannel.write(buf);
			db = null;
		}
		
		//	当前的pos就是Index的起始地址，写入Index
		int indexPos = pos;
		ByteBuffer indexBuf = ib.encode();
		pos += indexBuf.capacity();
		int indexLen = indexBuf.capacity();
		fileChannel.write(indexBuf);
		
		//	当前的pos就是Meta的起始地址，写入Meta
		int metaPos = pos;
		MetaBlock mb = new MetaBlock(size, blockCapacity, bf);
		ByteBuffer metaBuf = mb.encode();
		pos += metaBuf.capacity();
		int metaLen = metaBuf.capacity();
		fileChannel.write(metaBuf);
		
		//	写入文件尾
		TailBlock tb = new TailBlock(indexPos, indexLen, metaPos, metaLen);
		ByteBuffer tailBuf = tb.encode();
		fileChannel.write(tailBuf);
		
		//	关闭文件
		fileChannel.force(true);
		fileChannel.close();
	}
	
	
	/**
	 * 	检测目录是否存在
	 */
	protected void checkLevelDir(LSMConfig config, int level) {
		String dir = String.format("%s/%s%d", config.ROOT_DIR, config.levelPrefix, level);
		FileTools.tryMakeDir(dir);
	}
}
